package jmars.mobile.worker;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import jmars.mobile.common.Investigator;
import jmars.mobile.datainvestigation.Summary;

/**
 * select [group columns],count(distinct [target columns]) as col1 from data group by [group columns] order by col1
 * @author Jan
 *
 */
public class GroupDistinctCount implements Investigator {

	@Override
	public Long compute(File[] files, FileWriter writer) throws IOException {
		Long total = 0L;
		FileInputStream fis;
		BufferedReader reader;
		String row;
		String[] cols;
		Map<String,Long> distinct=new HashMap<String,Long>();
		Map<String,Set<String>> links=new HashMap<String,Set<String>>();
		Set<String> link;
		String groupKey,targetKey;
		DecimalFormat df = new DecimalFormat();
		df.applyPattern("0,000");
		for(File inputFile:files){
			fis = new FileInputStream(inputFile);
			reader = new BufferedReader(new InputStreamReader(fis));
			while((row=reader.readLine())!=null){
				cols=row.split(Summary.columnSeperator);
				if(Summary.filterRow(cols,true,true)) continue;
				groupKey="";
				targetKey="";
				for(Integer idx:Summary.groupColumnIdx){
					if(idx<cols.length) {
						groupKey=groupKey+"$_$"+cols[idx]+"$_$";
					}
					else groupKey=groupKey+"$_$$_$";
				}
				for(Integer idx:Summary.targetColumnIdx){
					if(idx<cols.length) {
						targetKey=targetKey+"$_$"+(cols[idx].lastIndexOf(".")==-1?cols[idx]:cols[idx].substring(0, cols[idx].lastIndexOf(".")))+"$_$";
					}
					else targetKey=targetKey+"$_$$_$";
				}

				if(!links.containsKey(groupKey)){
					distinct.put(groupKey,1L);
					link = new HashSet<String>();
					link.add(targetKey);
					links.put(groupKey, link);
				}else{
					link = links.get(groupKey);
					if(!link.contains(targetKey)){
						distinct.put(groupKey, distinct.get(groupKey)+1);
						link.add(targetKey);
					}
				}
				
				total++;
			}
			reader.close();
			fis.close();
			System.out.println(inputFile.getName()+" finished at "+total);
		}
		writer.write("%"+Summary.groupColumns+":"+df.format(distinct.size())+"\r\n");
		writer.write("%total:"+df.format(total)+"\r\n");
		if(distinct.size()<=1000000){
			ArrayList<Entry<String,Long>> l = new ArrayList<Entry<String,Long>>(distinct.entrySet());    
	          
	        Collections.sort(l, new Comparator<Map.Entry<String, Long>>() {    
	            public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {    
	                return (int) (o2.getValue() - o1.getValue());    
	            }    
	        });  

			writer.write("%*****************details******************\r\n%");
			for(Integer idx:Summary.groupColumnIdx){
				writer.write(Summary.columns[idx]+",");
			}
			writer.write("count_of_");
			for(Integer idx:Summary.targetColumnIdx){
				writer.write("_"+Summary.columns[idx]);
			}
			writer.write("\r\n");
	        for(Entry<String,Long> e : l) {
	            writer.write(e.getKey().replaceAll("\\$_\\$\\$_\\$", ",").replaceAll("\\$_\\$","")+","+e.getValue()+"\r\n");
	        }
		}
		return total;
	}

}
